package com.sjk.flink.functions;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.sjk.flink.correct.IfCorrectRule;
import com.sjk.flink.correct.rules.UnifyDateTimeRuler;
import com.sjk.flink.exeception.BaseException;
import com.sjk.flink.exeception.ErrorCode;
import com.sjk.flink.json.KeyConverter;
import com.sjk.flink.pojo.ECTTY;
import com.sjk.flink.pojo.OutPutPojo;
import com.sjk.flink.util.ExecptionUtil;
import com.sjk.flink.util.JsonSchemaUtil;
import com.sjk.flink.util.MessyCodeCheck;
import com.sjk.flink.util.ObjectMapperUtil;
import networknt.schema.utils.StringUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.TimeZone;

public class TimelinessProcessFunction extends ProcessFunction<ConsumerRecord<String, String>, ConsumerRecord<String, String>> {

    private static final Logger log = LoggerFactory.getLogger(JsonExpireProcessFunction.class);

    private final ParameterTool parameter;
    private final OutputTag<OutPutPojo> commonInvalidJsonTag;
    //private final OutputTag<OutPutPojo> expireJsonTag;

    private final OutputTag<OutPutPojo> notExpireJsonTag;


    private transient ObjectMapper objectMapper;

    private transient ParameterTool parameterTool;


    @Override
    public void open(Configuration parameters) {
        objectMapper = ObjectMapperUtil.getObjectMapper();
        parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
    }

    public TimelinessProcessFunction(ParameterTool parameter,OutputTag<OutPutPojo> notExpireJsonTag, OutputTag<OutPutPojo> commonInvalidJsonTag) {
        //this.expireJsonTag = expireJsonTag;
        this.notExpireJsonTag = notExpireJsonTag;
        this.commonInvalidJsonTag = commonInvalidJsonTag;
        this.parameter = parameter;
    }

    @Override
    public void processElement(ConsumerRecord<String, String> stringStringConsumerRecord, ProcessFunction<ConsumerRecord<String, String>, ConsumerRecord<String, String>>.Context ctx, Collector<ConsumerRecord<String, String>> collector) throws Exception {
        String value = stringStringConsumerRecord.value();
        String topic = stringStringConsumerRecord.topic();
        //log.error("消息主体:                           " + value);
        //log.info("来源topic:                  " + topic);
        int clayer_partition = stringStringConsumerRecord.partition();
        long clayer_offset = stringStringConsumerRecord.offset();
        Long clayer_ts = stringStringConsumerRecord.timestamp();

        String zxdm = parameterTool.get("zxdm");
        //log.info("专项代码                 " + zxdm);
        //log.info("过期标志                 " + parameter.getBoolean(topic + ".verify.flag", false));
        //log.info("data.field                 " + parameter.get(topic + ".data.field"));
        //log.info("compare.field                 " + parameter.get(topic + ".compare.field"));
        //获取表名 去除ODS_之前的字符
        String tableName = topic.substring(topic.indexOf("ODS_") + 4).trim();
        //log.info("tableName                      " + tableName);
        String topicsscsdm = "";

        OutPutPojo outPutPojo = new OutPutPojo();
        outPutPojo.setTableName(tableName);
        outPutPojo.setTopic(topic);
        outPutPojo.setClayer_partition(clayer_partition);
        outPutPojo.setClayer_offset(clayer_offset);
        outPutPojo.setClayer_ts(clayer_ts);
        //城市分区
        int city_partition_id = 0;
        //城市偏移量
        long city_offset_id = 0;
        //时间戳
        String city_ts = "";

        try {
            JsonNode readTree = objectMapper.readTree(value);
            JsonNode sscsdmJsonNode = readTree.get("sscsdm");
            //城市代码
            topicsscsdm = sscsdmJsonNode.asText();
            //抽取时间
            JsonNode extract_timeJsonNode = readTree.get("extract_time");
            String extract_time = extract_timeJsonNode.asText();
            //城市分区
            JsonNode city_partition_idJsonNode = readTree.get("city_partition_id");
            if (city_partition_idJsonNode != null) {
                city_partition_id = new Integer(city_partition_idJsonNode.asText());
            }
            //城市偏移量
            JsonNode city_offset_idJsonNode = readTree.get("city_offset_id");
            if (city_offset_idJsonNode != null) {
                city_offset_id = new Long(city_offset_idJsonNode.asText());
            }
            //时间戳
            JsonNode city_tsJsonNode = readTree.get("city_ts");
            if (city_tsJsonNode != null) {
                city_ts = city_tsJsonNode.asText();
            }
            //过期校验标志
            //boolean verify = parameter.getBoolean(topic + ".verify.flag", false);

            JsonNode recordJsonNode = readTree.get("record");
            String srcData1 = recordJsonNode.asText();
            if (MessyCodeCheck.isMessyCode(srcData1)) {
                //乱码
                throw new BaseException(ErrorCode.CODE_E00004);
            }
            //获得record数据
            JsonNode jsonNode = objectMapper.readTree(srcData1);

            //不做数据校准及质量分析直接到标准库
            boolean isObj = jsonNode.isObject();
            boolean isArray = jsonNode.isArray();
            if (isArray) {
                for (JsonNode node1 : jsonNode) {
                    handleJsonNode(ctx,outPutPojo,node1,city_ts,topicsscsdm,extract_time,city_partition_id,city_offset_id,clayer_partition,clayer_offset,clayer_ts);
                }
            } else if (isObj) {
                handleJsonNode(ctx,outPutPojo,jsonNode,city_ts,topicsscsdm,extract_time,city_partition_id,city_offset_id,clayer_partition,clayer_offset,clayer_ts);
            } else {
                outPutPojo.setJson(analysisExecption(topic, topicsscsdm, value, "000000", "不是有效的json数据", city_partition_id, city_offset_id, city_ts, clayer_partition, clayer_offset, clayer_ts));
                //不是有效的json
                //不是有效的 JSON，发送到无效数据的侧输出
                ctx.output(commonInvalidJsonTag, outPutPojo);
            }

        } catch (Exception e) {
            e.getStackTrace();
            String errorCode = e.getClass().getName();
            if (e instanceof BaseException) {
                errorCode = ((BaseException) e).getCode();
            }
            log.error("系统错误:" + ExecptionUtil.getStackTraceMsg(e));
            //城市代码可能为空
            outPutPojo.setJson(analysisExecption(topic, topicsscsdm, value, errorCode, "解析错误:" + ExecptionUtil.getStackTraceMsg(e), city_partition_id, city_offset_id, city_ts, clayer_partition, clayer_offset, clayer_ts));
            //解析 JSON 出错，发送到无效数据的侧输出
            ctx.output(commonInvalidJsonTag, outPutPojo);
        }
    }

    /**
     * 添加字符串
     *
     * @param ctx
     * @param outPutPojo
     * @param node
     * @param city_ts
     * @param topicsscsdm
     * @param extract_time
     * @param city_partition_id
     * @param city_offset_id
     * @param clayer_partition
     * @param clayer_offset
     * @param clayer_ts
     * @throws Exception
     */
    private void handleJsonNode( ProcessFunction<ConsumerRecord<String, String>, ConsumerRecord<String, String>>.Context ctx,OutPutPojo outPutPojo,JsonNode node,String city_ts,String topicsscsdm,String extract_time, int city_partition_id,long city_offset_id,int clayer_partition,long clayer_offset,long clayer_ts)throws Exception{
        //转为小写
        JsonNode lowerCaseNode = KeyConverter.convertKeysToLowercase(node);
        //补充数据
        ObjectNode plugJsonNode = objectMapper.createObjectNode();
        // 将原始的JsonNode内容复制到ObjectNode中
        plugJsonNode.setAll((ObjectNode)lowerCaseNode);
        // 把城市时间添加补充到用户数据中
        plugJsonNode.put("city_ts", city_ts);

        String arrayJsonStr = objectMapper.writeValueAsString(plugJsonNode);
        outPutPojo.setJson(packageJson(topicsscsdm, extract_time, arrayJsonStr, city_partition_id, city_offset_id, city_ts, clayer_partition, clayer_offset, clayer_ts));
        ctx.output(notExpireJsonTag, outPutPojo);
    }

    /**
     * 封装成功返回的JSON
     *
     * @param sscddm
     * @param extract_time
     */
    private String packageJson(String sscddm, String extract_time, String scrData, int city_partition_id, long city_offset_id, String city_ts, int clayer_partition, long clayer_offset, Long clayer_ts) throws
            JsonProcessingException {
        ObjectNode readTree = objectMapper.createObjectNode();
        //ObjectNode readTree = (ObjectNode)objectMapper.readTree(scrData);
        //把 city_ts加入到scrData中
        readTree.put("record", scrData);
        readTree.put("sscsdm", sscddm);
        readTree.put("extract_time", extract_time);
        readTree.put("operate_time", JsonSchemaUtil.getNowTime("yyyy-MM-dd HH:mm:ss"));
        readTree.put("city_partition_id", city_partition_id);
        readTree.put("city_offset_id", city_offset_id);
        readTree.put("city_ts", city_ts);
        readTree.put("clayer_partition", clayer_partition);
        readTree.put("clayer_offset", clayer_offset);
        if (clayer_ts == null) {
            readTree.put("clayer_ts", "");
        } else {
            readTree.put("clayer_ts", formatDate(clayer_ts));
        }
        return objectMapper.writeValueAsString(readTree);
    }

    private String formatDate(long s) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        // 设置时区为东八区（北京）
        TimeZone timeZone = TimeZone.getTimeZone("GMT+8");
        sdf.setTimeZone(timeZone);

        // 获取当前时间
        Date date = new Date(s);

        // 将日期按照指定的格式转换成字符串
        return sdf.format(date);
    }

    /**
     * 解析及系统错误分析
     *
     * @param topic
     * @param cs
     * @param srcData
     * @return
     */
    private String analysisExecption(String topic, String cs, String srcData, String code, String message, int city_partition_id, long city_offset_id, String city_ts, int clayer_partition, long clayer_offset, Long clayer_ts) throws JsonProcessingException {
        ECTTY ectty = new ECTTY();
        /** topic名称 **/
        ectty.setTopicmc(topic);

        /** 城市 **/
        ectty.setCs(cs);

        /** 原始数据 **/
        ectty.setScrData(srcData);

        /** 错误代码 **/
        ectty.setErrorCode(code);
        /** 错误信息 **/
        ectty.setErrorMessage(message);
        ectty.setCity_partition_id(city_partition_id);
        ectty.setCity_offset_id(city_offset_id);
        ectty.setCity_ts(city_ts);

        ectty.setClayer_partition(clayer_partition);
        ectty.setClayer_offset(clayer_offset);
        if (clayer_ts == null) {
            ectty.setClayer_ts("");
        } else {
            ectty.setClayer_ts(formatDate(clayer_ts));
        }

        ectty.setOperate_time(JsonSchemaUtil.getNowTime("yyyy-MM-dd HH:mm:ss"));
        String json = objectMapper.writeValueAsString(ectty);
        return json;
    }

}
